package com.bff.gaia.mix.api.xjoin2.core.table;

import com.bff.gaia.mix.api.xjoin2.core.side.SideTableInfo;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Lists;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Maps;

import java.io.Serializable;
import java.util.List;
import java.util.Map;

public abstract class XjoinInfo implements Serializable {
	protected StreamTableInfo streamTableInfo;
	protected SideTableInfo sideTableInfo;
	protected JoinInfo joinInfo;
	protected List<FieldInfo> outFieldInfoList;

	protected List<Integer> streamTableKeyFieldPos = Lists.newArrayList();

	protected List<Integer> sideTableKeyFieldPos = Lists.newArrayList();
	protected List<String> sideTableKeyFieldVal = Lists.newArrayList();

	protected String sqlQuery = "";
	protected String sideSelectFields = "";

	//key:Returns the value of the position, value: the ref field index​in the input table
	protected Map<Integer, Integer> streamTableOutFieldIndex = Maps.newHashMap();

	//key:Returns the value of the position, value:  the ref field index​in the side table
	protected Map<Integer, String> sideTableOutFieldIndex = Maps.newHashMap();

	//key:Returns the value of the position, value:  the ref field index​in the side table
	protected Map<Integer, String> sideTableInFieldIndex = Maps.newHashMap();


	protected int outFieldSize;

	public XjoinInfo(StreamTableInfo streamTableInfo, SideTableInfo sideTableInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList) {
		this.streamTableInfo = streamTableInfo;
		this.sideTableInfo = sideTableInfo;
		this.joinInfo = joinInfo;
		this.outFieldInfoList = outFieldInfoList;
		outFieldSize = outFieldInfoList.size();
		parseSelectFields(joinInfo);
		buildQuery(joinInfo, sideTableInfo);
	}

	public void parseSelectFields(JoinInfo joinInfo) {
		String sideTableName = joinInfo.getSideTableName();
		String streamTableName = joinInfo.getStreamTableName();
		List<String> fields = Lists.newArrayList();

		int inFieldIndex = 0;
		int outFieldIndex = 0;
		for (FieldInfo fieldInfo : this.outFieldInfoList) {
			if (fieldInfo.getTable().equalsIgnoreCase(streamTableName)) {
				if (fieldInfo.isJoinKey()) {
					streamTableKeyFieldPos.add(fieldInfo.getFieldPos());
				}
				streamTableOutFieldIndex.put(outFieldIndex, fieldInfo.getFieldPos());
				outFieldIndex++;
			} else if (fieldInfo.getTable().equalsIgnoreCase(sideTableName)) {
				fields.add(fieldInfo.getFieldName());
				sideTableInFieldIndex.put(inFieldIndex++, fieldInfo.getFieldName());
				if (fieldInfo.isJoinKey()) {
					sideTableKeyFieldPos.add(fieldInfo.getFieldPos());
					sideTableKeyFieldVal.add(fieldInfo.getFieldName());
					outFieldSize--;
				} else {
					sideTableOutFieldIndex.put(outFieldIndex, fieldInfo.getFieldName());
					outFieldIndex++;
				}
			} else {
				throw new RuntimeException("unknown table " + fieldInfo.getTable());
			}
		}

		if (fields.size() == 0) {
			throw new RuntimeException("select non field from table " + sideTableName);
		}

		sideSelectFields = String.join(",", fields);
	}

	public abstract void buildQuery(JoinInfo joinInfo, SideTableInfo sideTableInfo);

	public StreamTableInfo getStreamTableInfo() {
		return streamTableInfo;
	}

	public SideTableInfo getSideTableInfo() {
		return sideTableInfo;
	}

	public JoinInfo getJoinInfo() {
		return joinInfo;
	}

	public List<FieldInfo> getOutFieldInfoList() {
		return outFieldInfoList;
	}

	public List<Integer> getStreamTableKeyFieldPos() {
		return streamTableKeyFieldPos;
	}

	public List<Integer> getSideTableKeyFieldPos() {
		return sideTableKeyFieldPos;
	}

	public String getSqlQuery() {
		return sqlQuery;
	}

	public String getSideSelectFields() {
		return sideSelectFields;
	}

	public Map<Integer, Integer> getStreamTableOutFieldIndex() {
		return streamTableOutFieldIndex;
	}

	public Map<Integer, String> getSideTableOutFieldIndex() {
		return sideTableOutFieldIndex;
	}

	public int getOutFieldSize() {
		return outFieldSize;
	}

	public List<String> getSideTableKeyFieldVal() {
		return sideTableKeyFieldVal;
	}

	public Map<Integer, String> getSideTableInFieldIndex() {
		return sideTableInFieldIndex;
	}
}